package com.atguigu.Driver;

import com.atguigu.Bean.OrderBean;
import com.atguigu.Mapper.GroupMapper;
import com.atguigu.Partition.GroupComparator;
import com.atguigu.Partition.GroupPartitioner;
import com.atguigu.Reducer.GroupReducer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class GroupDriver extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception {
        //获取Job对象
        Job job = Job.getInstance(super.getConf(), "myGroupComparator");
        //输入数据，设置输入路径
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.setInputPaths(job, new Path("D:\\SysProject\\HdfsClientDemo\\src\\main\\resources\\Input\\GroupCompare.txt"));

        //自定义Map逻辑
        job.setMapperClass(GroupMapper.class);
        //设置k2,v2输出类型
        job.setMapOutputKeyClass(OrderBean.class);
        job.setMapOutputValueClass(NullWritable.class);

        //自定义Partition逻辑
        job.setPartitionerClass(GroupPartitioner.class);

        //自定义分组逻辑
        job.setGroupingComparatorClass(GroupComparator.class);

        //自定义reduce逻辑
        job.setReducerClass(GroupReducer.class);
        //设置k3,v3输出类型
        job.setOutputKeyClass(OrderBean.class);
        job.setOutputValueClass(NullWritable.class);

        //输出数据，设置输出路径
        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job, new Path("D:\\SysProject\\HdfsClientDemo\\src\\main\\resources\\groupOut"));

        //提交任务至集群
        boolean b = job.waitForCompletion(true);
        return b ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        int run = ToolRunner.run(new Configuration(), new GroupDriver(), args);
        System.exit(run);
    }
}
